package com.test;

import com.temp.test.entity.User;
import com.temp.test.service.UserService;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: Gallrax
 * @Description:
 * @Date: 2018/7/18
 */
public class BatchInsertUtil {

    public static final Integer EXECUTE_COUNT = 100000;
    private static UserService userService;

    public static UserService getUserService() {
        return userService;
    }

    public static void testSingle(UserService userService) {
        BatchInsertUtil.userService = userService;
        //执行线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        executorService.execute(() -> System.out.println("-"));
    }

    public static void testMulti(UserService userService) {
        BatchInsertUtil.userService = userService;
        //执行线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //执行队列
        LinkedBlockingQueue<User> linkedBlockingQueue = new LinkedBlockingQueue<>();
        //执行线程个数
        Integer threadCount = 5;
        //批量增加个数
//        CountDownLatch count = new CountDownLatch(10000);

        //生产者
        List<Runnable> producerThreads = getProducerThreads(threadCount, linkedBlockingQueue);
        for (Runnable runnable : producerThreads) {
            executorService.execute(runnable);
        }
        //消费者
        List<Runnable> consumerThreads = getConsumerThreads(5, linkedBlockingQueue);
        //执行任务
        excuteMultiThread(executorService, producerThreads);
        excuteMultiThread(executorService, consumerThreads);
    }

    private static void excuteMultiThread(Executor executor, List<Runnable> runnables) {
        Assert.notNull(executor, "executor must not null");
        Assert.notEmpty(runnables, " threads must not null");
        for (Runnable runnable : runnables) {
            executor.execute(runnable);
        }
    }

    private static List<Runnable> getProducerThreads(int size, BlockingQueue blockingQueue) {
        List<Runnable> list = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            list.add(new ProducerThread(blockingQueue));
        }
        return list;
    }

    private static List<Runnable> getConsumerThreads(int size, BlockingQueue blockingQueue) {
        List<Runnable> list = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            list.add(new ConsumerThread(blockingQueue));
        }
        return list;
    }

    public static User buildUser(Integer i) {
        String tempStr = "temp" + i;
        User user = new User();
        user.setName(tempStr);
        user.setIntro(tempStr);
        user.setAddress(tempStr);
        user.setSex(0);
        user.setPhone("13811111111");
        return user;
    }


}

class ProducerThread implements Runnable {

    private boolean isRunning = true;//多线程执行状态(停止则全部停止)
    private BlockingQueue<User> queue;//消息队列
    private static AtomicInteger count = new AtomicInteger(0);//线程原子操作当前操作数
    private static final int SLEEPTime = 1000;//睡眠时长

    public ProducerThread(BlockingQueue<User> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (isRunning) {
            //当操作任务剩余数为0时，停止执行
            int i = count.getAndIncrement();
            if (i > BatchInsertUtil.EXECUTE_COUNT) isRunning = false;
            //不采用count.get(),回到值cas，影响效率
            User user = BatchInsertUtil.buildUser(i);
            System.out.println(" ProdcerThread is Running, currentThread : " + Thread.currentThread().toString() + " i: " + i + " currentTime : " + System.currentTimeMillis() + " user : " + user.toString());
            try {
                queue.put(user);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

class ConsumerThread implements Runnable {

    private boolean isRunning = true;//多线程执行状态
    private BlockingQueue<User> queue;//消息队列
    private static AtomicInteger count = new AtomicInteger(0);//多线程原子操作总数
    private static final int SLEEPTime = 1000;//睡眠时长

    public ConsumerThread(BlockingQueue<User> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        //当操作任务剩余数为0时，停止执行
        while (isRunning) {
            try {
                User user = queue.take();
                UserService userService = BatchInsertUtil.getUserService();
                userService.insert(user);
                //执行user插入操作
                int i = count.getAndIncrement();
                System.out.println(" ConsumerThread is Running, currentThread : " + Thread.currentThread().toString() + " i " + i + " currentTime : " + System.currentTimeMillis() + " user : " + user.toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
